package com.shujia.sql

import org.apache.spark.sql.expressions.Window
import org.apache.spark.sql.{DataFrame, SaveMode, SparkSession}

object Code13Burks {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
//      .master("local")
      .appName("dsl")
      .config("spark.sql.shuffle.partitions", "3")
      .getOrCreate()

    /**
     *
     * spark-submit提交命令：
     *  spark-submit  --class   com.shujia.sql.Code13Burks  --master yarn --executor-memory 512m --total-executor-cores 1 --num-executors 3  spark_code-1.0.jar
     *
     */


    val burkDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", ",")
      // 1500100001,施笑槐,22,女,文科六班
      .schema("burk String,year int,tsl01 int,tsl02 int,tsl03 int,tsl04 int,tsl05 int,tsl06 int,tsl07 int,tsl08 int,tsl09 int,tsl10 int,tsl11 int,tsl12 int")
//      .load("spark_code/data/burks.csv")
//      .load("/data/burks.csv")
      .load(" /usr/local/soft/sparkJob/burks.csv")

    import org.apache.spark.sql.functions._
    import spark.implicits._

    //    burkDF
    //      //      .select($"burk",$"year",explode(array($"tsl01",$"tsl02",$"tsl03",$"tsl04",$"tsl05",$"tsl06",$"tsl07",$"tsl08",$"tsl09",$"tsl10",$"tsl11",$"tsl12")))
    //      .select($"burk", $"year", explode(array(
    //        array(expr("1"), $"tsl01")
    //        , array(expr("2"), $"tsl02")
    //        , array(expr("3"), $"tsl03")
    //        , array(expr("4"), $"tsl04")
    //        , array(expr("5"), $"tsl05")
    //        , array(expr("6"), $"tsl06")
    //        , array(expr("7"), $"tsl07")
    //        , array(expr("8"), $"tsl08")
    //        , array(expr("9"), $"tsl09")
    //        , array(expr("10"), $"tsl10")
    //        , array(expr("11"), $"tsl11")
    //        , array(expr("12"), $"tsl12"))))
    //      .select($"burk", $"year",expr("col[0]") as "mon",expr("col[1]") as "salary")
    //      .show()

    val transformData: DataFrame = burkDF
      .select($"burk", $"year", posexplode(array($"tsl01", $"tsl02", $"tsl03", $"tsl04", $"tsl05", $"tsl06", $"tsl07", $"tsl08", $"tsl09", $"tsl10", $"tsl11", $"tsl12")))
      .select($"burk", $"year", $"pos" + 1 as "mon", $"col" as "salary")

    transformData
      // 统计每个公司每年按月累计收入 输出结果：公司代码,年度,月份,当月收入,累计收入
      .withColumn("sum_salary", sum($"salary") over (Window.partitionBy($"burk", $"year").orderBy("mon")))
    //      .show()

    // 统计每个公司当月比上年同期增长率  公司代码,年度,月度,去年当月收入,当月收入,增长率
    transformData
      .withColumn("qn_mon_salary", lag("salary", 1,0) over Window.partitionBy($"burk",$"mon").orderBy($"mon", $"year"))
      .withColumn("increase", round(($"salary" - $"qn_mon_salary") / $"qn_mon_salary",6))
      .write
      .format("csv")
      .option("sep",",")
      .mode(SaveMode.Overwrite)
      .save("/output/spark/burks")



  }
}
